package com.auto.storm;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import lombok.extern.java.Log;

import org.apache.storm.shade.com.google.common.collect.Maps;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.auto.redisMq.JedisUtil;

/** 
* @ClassName: SocketCountBolt 
* @Description: 分产品计算数据流
* @author hcw 
* @date 2016年12月27日 上午10:33:07 
*  
*/
@Log
public class SocketCountBolt extends BaseRichBolt{
	private OutputCollector _collector;
	
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this._collector =collector;
	}
	
	//blot计算单元,把相同的数据合成流放出去;  
	@Override
	public void execute(Tuple input) {
		  String productID = input.getString(0);
		  String productName = input.getStringByField("productValue");// 通过Field名称取出对应字段的数据
		  log.info("通道最终: productId=" + productID+",最终产品值:"+productName);
		  JedisUtil.lpush(productID.getBytes(), productName.getBytes());
		  //成功通知
		  _collector.ack(input);
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}
}
